package com.tyjt.common.mqtt.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class MqttCallBack implements MqttCallbackExtended , CommandLineRunner {

    private static MqttClient mqttClient;

    public MqttCallBack(MqttClient mqttClient) {
        MqttCallBack.mqttClient = mqttClient;
    }

    public static void sub() {
        if (mqttClient.isConnected()) {
            try {
                if (DataHandlerManager.getINSTANCE().topics().length != 0){
                    //  订阅数据
                    mqttClient.subscribe(DataHandlerManager.getINSTANCE().topics());
                    log.info("订阅topic:{} 成功！", Arrays.asList(DataHandlerManager.getINSTANCE().topics()));
                }
            } catch (MqttException e) {
                log.error("订阅topic:{} 失败！", Arrays.asList(DataHandlerManager.getINSTANCE().topics()));
                throw new RuntimeException(e);
            }
        } else {
            log.error("mqtt 连接失败");
        }
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("连接 EMQX 成功： reconnect  {}  severURI  {}", reconnect, serverURI);
        if (reconnect) {
            new Thread(MqttCallBack::sub).start();
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.warn("EMQX 连接断开！", cause);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        Set<String> patterns = DataHandlerManager.getINSTANCE().patternSet();

        for (String pattern : patterns) {
            boolean flag = MqttTopic.isMatched(pattern, topic);
            if (flag) {
                callbackExecutor.submit(()->{
                    try {
                        DataHandlerManager.getINSTANCE().getHandler(pattern).handle(topic, message);
                    } catch (Exception e) {
                        log.error("",e);
                    }
                });
            }
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
//        log.info("deliveryComplete...");
    }


    @Override
    public void run(String... args) {
        sub();
    }


    // 1. 定义线程池（核心参数根据机器配置调整）
    int cpuCores = Runtime.getRuntime().availableProcessors();
    int corePoolSize = cpuCores * 2;          // 计算密集型可适当减少，IO密集型可增加
    int maxPoolSize = cpuCores * 4;           // 突发流量时扩容
    int queueCapacity = 10000;                // 根据内存调整，避免OOM
    long keepAliveTime = 60L;                 // 空闲线程存活时间（秒）

    ExecutorService callbackExecutor = new ThreadPoolExecutor(
            corePoolSize,
            maxPoolSize,
            keepAliveTime,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(queueCapacity), // 有界队列
            new ThreadPoolExecutor.DiscardOldestPolicy()   // 队列满时直接丢弃最旧任务（根据需求选策略）
    );
}
